package com.djf.flinkcdc.controller;

import com.djf.flinkcdc.config.FlinkCdcProperties;
import com.djf.flinkcdc.service.CdcSyncService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.HashMap;
import java.util.Map;

/**
 * CDC 同步管理控制器
 * 提供数据同步、一致性和迁移的管理接口
 */
@Slf4j
@RestController
@RequestMapping("/api/cdc")
public class CdcController {

    @Autowired
    private CdcSyncService cdcSyncService;
    
    @Autowired
    private FlinkCdcProperties properties;

    /**
     * 健康检查
     */
    @GetMapping("/health")
    public Map<String, Object> health() {
        Map<String, Object> result = new HashMap<>();
        result.put("status", "UP");
        result.put("service", "Flink CDC Sync Service");
        result.put("running", cdcSyncService.isRunning());
        result.put("timestamp", System.currentTimeMillis());
        return result;
    }

    /**
     * 获取配置信息
     */
    @GetMapping("/config")
    public Map<String, Object> getConfig() {
        Map<String, Object> config = new HashMap<>();
        
        // MySQL 配置
        Map<String, Object> mysql = new HashMap<>();
        mysql.put("hostname", properties.getMysql().getHostname());
        mysql.put("port", properties.getMysql().getPort());
        mysql.put("database", properties.getMysql().getDatabaseList());
        mysql.put("tables", properties.getMysql().getTableList());
        mysql.put("startupMode", properties.getMysql().getStartupMode());
        mysql.put("serverId", properties.getMysql().getServerId());
        mysql.put("timeZone", properties.getMysql().getServerTimeZone());
        config.put("mysql", mysql);
        
        // Checkpoint 配置
        Map<String, Object> checkpoint = new HashMap<>();
        checkpoint.put("enabled", properties.getCheckpoint().getEnabled());
        checkpoint.put("interval", properties.getCheckpoint().getInterval());
        checkpoint.put("storage", properties.getCheckpoint().getStorage());
        config.put("checkpoint", checkpoint);
        
        return config;
    }

    /**
     * 获取同步状态
     */
    @GetMapping("/status")
    public Map<String, Object> getStatus() {
        Map<String, Object> status = new HashMap<>();
        status.put("running", cdcSyncService.isRunning());
        status.put("service", "CDC Data Sync");
        status.put("features", new String[]{"数据同步", "数据一致性", "数据迁移"});
        status.put("timestamp", System.currentTimeMillis());
        return status;
    }

    /**
     * 启动数据同步
     */
    @PostMapping("/start")
    public Map<String, Object> startSync() {
        Map<String, Object> result = new HashMap<>();
        
        try {
            if (cdcSyncService.isRunning()) {
                result.put("success", false);
                result.put("message", "同步任务已在运行中");
                return result;
            }
            
            cdcSyncService.startSync();
            
            result.put("success", true);
            result.put("message", "数据同步任务启动成功");
            result.put("features", new String[]{"数据同步", "数据一致性保证", "数据迁移"});
            result.put("timestamp", System.currentTimeMillis());
            
            log.info("✅ 通过API启动数据同步任务");
            
        } catch (Exception e) {
            result.put("success", false);
            result.put("message", "启动失败: " + e.getMessage());
            log.error("❌ API启动同步任务失败", e);
        }
        
        return result;
    }

    /**
     * 停止数据同步
     */
    @PostMapping("/stop")
    public Map<String, Object> stopSync() {
        Map<String, Object> result = new HashMap<>();
        
        try {
            if (!cdcSyncService.isRunning()) {
                result.put("success", false);
                result.put("message", "同步任务未在运行");
                return result;
            }
            
            cdcSyncService.stopSync();
            
            result.put("success", true);
            result.put("message", "数据同步任务停止成功");
            result.put("timestamp", System.currentTimeMillis());
            
            log.info("✅ 通过API停止数据同步任务");
            
        } catch (Exception e) {
            result.put("success", false);
            result.put("message", "停止失败: " + e.getMessage());
            log.error("❌ API停止同步任务失败", e);
        }
        
        return result;
    }

    /**
     * 重启数据同步
     */
    @PostMapping("/restart")
    public Map<String, Object> restartSync() {
        Map<String, Object> result = new HashMap<>();
        
        try {
            log.info("🔄 重启数据同步任务...");
            
            // 先停止
            if (cdcSyncService.isRunning()) {
                cdcSyncService.stopSync();
                Thread.sleep(2000); // 等待停止
            }
            
            // 再启动
            cdcSyncService.startSync();
            
            result.put("success", true);
            result.put("message", "数据同步任务重启成功");
            result.put("timestamp", System.currentTimeMillis());
            
            log.info("✅ 通过API重启数据同步任务");
            
        } catch (Exception e) {
            result.put("success", false);
            result.put("message", "重启失败: " + e.getMessage());
            log.error("❌ API重启同步任务失败", e);
        }
        
        return result;
    }

    /**
     * 获取功能说明
     */
    @GetMapping("/features")
    public Map<String, Object> getFeatures() {
        Map<String, Object> features = new HashMap<>();
        
        // 替换不兼容的Map.of方法为传统Map创建方式
        Map<String, Object> dataSyncMap = new HashMap<>();
        dataSyncMap.put("name", "数据同步");
        dataSyncMap.put("description", "实时捕获MySQL数据变更，支持全量和增量同步");
        dataSyncMap.put("modes", new String[]{"initial", "latest-offset", "earliest-offset"});
        features.put("dataSync", dataSyncMap);

        Map<String, Object> dataConsistencyMap = new HashMap<>();
        dataConsistencyMap.put("name", "数据一致性");
        dataConsistencyMap.put("description", "通过Flink的Exactly-Once语义保证数据一致性");
        dataConsistencyMap.put("features", new String[]{"Checkpoint机制", "故障恢复", "重复数据检测"});
        features.put("dataConsistency", dataConsistencyMap);

        Map<String, Object> dataMigrationMap = new HashMap<>();
        dataMigrationMap.put("name", "数据迁移");
        dataMigrationMap.put("description", "支持数据库间的数据迁移和同步");
        dataMigrationMap.put("capabilities", new String[]{"表结构变更感知", "数据转换", "目标系统适配"});
        features.put("dataMigration", dataMigrationMap);
        
        return features;
    }
}